/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package com.eluup.flume.sink.elasticsearch;

import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;

import java.io.IOException;
import java.lang.reflect.Type;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.TimeZone;

import org.apache.commons.lang.ObjectUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.time.FastDateFormat;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.conf.ComponentConfiguration;
import org.apache.flume.formatter.output.BucketPath;
import org.elasticsearch.common.xcontent.XContentBuilder;

import com.google.common.collect.Maps;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.LongSerializationPolicy;
import com.google.gson.reflect.TypeToken;

/**
 * Serialize flume events into the same format LogStash uses</p>
 * <p>
 * This can be used to send events to ElasticSearch and use clients such as
 * Kabana which expect Logstash formated indexes
 * <p>
 * 
 * <pre>
 * {
 *    "@timestamp": "2010-12-21T21:48:33.309258Z",
 *    "@tags": [ "array", "of", "tags" ],
 *    "@type": "string",
 *    "@source": "source of the event, usually a URL."
 *    "@source_host": ""
 *    "@source_path": ""
 *    "@fields":{
 *       # a set of fields for this event
 *       "user": "jordan",
 *       "command": "shutdown -r":
 *     }
 *     "other_field":"other_value"
 *   }
 * </pre>
 * <p>
 * If the following headers are present, they will map to the above logstash
 * output as long as the logstash fields are not already present.
 * </p>
 * <p>
 * 
 * <pre>
 *  timestamp: long -> @timestamp:Date
 *  host: String -> @source_host: String
 *  src_path: String -> @source_path: String
 *  type: String -> @type: String
 *  source: String -> @source: String
 * </pre>
 *
 * @see https
 *      ://github.com/logstash/logstash/wiki/logstash%27s-internal-message-
 *      format
 */
public class ElasticSearchScriptSerializer implements ElasticSearchEventSerializer {

	public static final String TIME_ZONE = "timeZone";

	public static final String DEFAULT_DATE_FORMAT = "yyyy-MM-dd'T'HH:mm:ss";
	public static final String DEFAULT_TIME_ZONE = "Asia/Shanghai";

	private FastDateFormat fastDateFormat = FastDateFormat.getInstance("yyyy-MM-dd'T'HH:mm:ss",
	        TimeZone.getTimeZone("Asia/Shanghai"));

	
	private StringBuffer script = new StringBuffer();
	private String primaryKey;
	private Long version;
	private Map<String, Object> params;
	

	public XContentBuilder getContentBuilder(Event event) throws IOException {
		version = 1L;
		primaryKey = "empty";
		script.setLength(0);
		params = new HashMap<String, Object>();
		
		XContentBuilder builder = jsonBuilder().startObject();
		appendBody(builder, event);
		appendHeaders(builder, event);
		builder.endObject();
		return builder;
	}
	public Long getVersion() {
		return version;
	}
	public String getPrimaryKey() {
		return primaryKey;
	}
	public String getScript() {
		if (script.length() > 0) {
			return script.substring(0, script.length()-1);
		} else {
			return null;
		}
	}

	public Map<String, Object> getParams() {
		return params;
	}
	private void appendBody(XContentBuilder builder, Event event) throws IOException {
		// ContentBuilderUtil.appendField(builder, "msg", body);
		Gson gson = new GsonBuilder()
			.setLongSerializationPolicy(LongSerializationPolicy.STRING)
			.create();
		Type type = new TypeToken<Map<String, Object>>() {}.getType();
		Map<String, Object> jsonHash = gson.fromJson(new String(event.getBody(), charset), type);
		// 这只版本号
		version = Long.valueOf(jsonHash.getOrDefault("version", "1").toString());
		if (jsonHash.containsKey("Update")) {
			String[] Update = (jsonHash.get("Update").toString() + "^" + "FAT").split("\\^");
			jsonHash.put("FAT", fastDateFormat.format(new Date()));
			jsonHash = getAppendBody(builder, jsonHash, Update, "Update");
			jsonHash.remove("Update");
		}
		if (jsonHash.containsKey("Append")) {
			String[] Append = jsonHash.get("Append").toString().split("\\^");
			jsonHash = getAppendBody(builder, jsonHash, Append,  "Append");
			jsonHash.remove("Append");
		}
		if (jsonHash.containsKey("Incr")) {
			String[] Incr = jsonHash.get("Incr").toString().split("\\^");
			jsonHash = getAppendBody(builder, jsonHash, Incr, "Incr");
			jsonHash.remove("Incr");
		}
		
		ContentBuilderUtil.appendField(builder, "@message", gson.toJson(jsonHash).getBytes());
	}

	private Map<String, Object> getAppendBody(XContentBuilder builder, Map<String, Object> jsonHash, String[] features, String type) throws IOException {
		for (String feature : features) {
			if(jsonHash.containsKey(feature) && !jsonHash.get(feature).equals("0") && !jsonHash.get(feature).equals("")) {
				final Object param;
				Object defValue = null;
				if (type.equals("Append")) {
					script.append("ctx._source."+feature+".add(params."+feature+");");
					if (jsonHash.get(feature) instanceof java.util.ArrayList) {
						param = jsonHash.get(feature);
					} else {
						param = parseString(jsonHash.get(feature));
					}
					defValue = new ArrayList<Object>(){{add(param);}};
				} else if (type.equals("Update")) {
					script.append("ctx._source."+feature+" = params."+feature+";");
					if (jsonHash.get(feature) instanceof java.util.ArrayList) {
						param = jsonHash.get(feature);
					} else {
						param = parseString(jsonHash.get(feature));
					}
					defValue = param;
				} else if (type.equals("Incr")) {
					script.append("ctx._source."+feature+" += params."+feature+";");
					param = Long.valueOf(parseString(jsonHash.get(feature)));
					defValue = param;
				} else {
					param = "";
					defValue = "";
				}
				params.put(feature, param);
				builder.field(feature, defValue);
				jsonHash.remove(feature);
			} else {
				if (type.equals("Append")) {
					builder.field(feature, new String[0]);
				} else if (type.equals("Update")) {
					builder.field(feature, "");
				} else if (type.equals("Incr")) {
					builder.field(feature, 0);
				} else {
					builder.field(feature, "");
				}
			}
		}
		return jsonHash;
	}
	private void appendHeaders(XContentBuilder builder, Event event) throws IOException {
		
		Map<String, String> headers = Maps.newHashMap(event.getHeaders());
		String timestamp = headers.get("timestamp");
		if (!StringUtils.isBlank(timestamp) && StringUtils.isBlank(headers.get("@timestamp"))) {
			long timestampMs = Long.parseLong(timestamp);
			builder.field("@timestamp", fastDateFormat.format(timestampMs));
		}
		
		String source = headers.get("source");
		if (!StringUtils.isBlank(source) && StringUtils.isBlank(headers.get("@source"))) {
			ContentBuilderUtil.appendField(builder, "@source", source.getBytes(charset));
		}

		String type = headers.get("type");
		if (!StringUtils.isBlank(type) && StringUtils.isBlank(headers.get("@type"))) {
			ContentBuilderUtil.appendField(builder, "@type", type.getBytes(charset));
		}

		String host = headers.get("host");
		if (!StringUtils.isBlank(host) && StringUtils.isBlank(headers.get("@source_host"))) {
			ContentBuilderUtil.appendField(builder, "@source_host", host.getBytes(charset));
		}

		String srcPath = headers.get("src_path");
		if (!StringUtils.isBlank(srcPath) && StringUtils.isBlank(headers.get("@source_path"))) {
			ContentBuilderUtil.appendField(builder, "@source_path", srcPath.getBytes(charset));
		}

		builder.startObject("@fields");
		for (Map.Entry<String, String> entry : headers.entrySet()) {
			byte[] val = entry.getValue().getBytes(charset);
			ContentBuilderUtil.appendField(builder, entry.getKey(), val);
			if (entry.getKey() == "") {
				
			}
		}
		builder.endObject();
		primaryKey = headers.get("key");
	}

	public void configure(Context context) {
		String timeZoneString = context.getString(TIME_ZONE);
		String dateFormatString = DEFAULT_DATE_FORMAT;
		if (StringUtils.isBlank(timeZoneString)) {
			timeZoneString = DEFAULT_TIME_ZONE;
		}
		fastDateFormat = FastDateFormat.getInstance(dateFormatString,
		        TimeZone.getTimeZone(timeZoneString));
	}

	public void configure(ComponentConfiguration conf) {
		// NO-OP...
	}
	public static String parseString(Object str){
        String returnValue = "";
        try {
            returnValue = new BigDecimal(String.valueOf(str)).toString();
            if(StringUtils.endsWith(returnValue, ".0")) {
                returnValue = StringUtils.removeEnd(returnValue, ".0");
            }
        } catch (Exception e) {
            returnValue = ObjectUtils.toString(str);
        }
        return returnValue;
    }
}
